// package: rewords-server-2024
// filename: cmgox
// author: diogoxiang@gmail.com
// time: 2024/7/31 上午10:51

package cmgox

import (
	"context"
	"fmt"

	"runtime"
	"strconv"
	"strings"
	"time"

	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/segmentio/fasthash/fnv1a"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"

	"rewords-server-2024/internal/global"
	"rewords-server-2024/internal/library/cache"
	"rewords-server-2024/internal/library/cfg"
	"rewords-server-2024/internal/library/xpool"
	"rewords-server-2024/utility/charset"
)

var nocacheList = g.Cfg().MustGet(context.Background(), "runtimes.nocacheList")

// generateCacheKey 生成缓存键
func generateCacheKey(coll string, matchStage bson.D, groupStage ...bson.D) string {
	h := fnv1a.Init64
	h = fnv1a.AddString64(h, coll)
	h = fnv1a.AddString64(h, runFuncName())
	h = fnv1a.AddString64(h, fmt.Sprintf("%v", matchStage))
	for _, stage := range groupStage {
		h = fnv1a.AddString64(h, fmt.Sprintf("%v", stage))
	}
	return fmt.Sprintf("%s_%s_%s", coll, runFuncName(), strconv.FormatUint(h, 10))
}

// 获取正在运行的函数名
func runFuncName() string {
	pc := make([]uintptr, 1)
	runtime.Callers(3, pc)
	f := runtime.FuncForPC(pc[0])
	funName := strings.ReplaceAll(f.Name(), "/", "_")
	return funName
}

// MongoPipeline 可以简化错误处理
func MongoPipeline(str string) (mongo.Pipeline, error) {
	var pipeline mongo.Pipeline
	trimmedStr := strings.TrimSpace(str)

	// 先尝试解析为pipeline，如果失败则尝试解析为单个文档
	if err := bson.UnmarshalExtJSON([]byte(trimmedStr), false, &pipeline); err != nil {
		var doc bson.D
		if err := bson.UnmarshalExtJSON([]byte(trimmedStr), false, &doc); err != nil {
			return nil, fmt.Errorf("failed to parse pipeline: %w", err)
		}
		return mongo.Pipeline{doc}, nil
	}
	return pipeline, nil
}

// aggregate is a helper function to perform the common aggregation logic
func aggregate(ctx context.Context, coll string, matchStage bson.D, groupStage ...bson.D) (*mongo.Cursor, error) {
	var (
		collection *mongo.Collection
		client     *mongo.Client
		opts       = options.AggregateOptions{}
		p          interface{}
	)

	p, err := xpool.GMongoPool.Get()
	if err != nil {
		return nil, fmt.Errorf("failed to get mongo connection from pool: %w", err)
	}
	defer xpool.GMongoPool.Put(p)

	client = p.(*mongo.Client)
	collection = client.Database(cfg.GConfig.Mongo.DbName).Collection(coll)

	sStage := mongo.Pipeline{matchStage}
	sStage = append(sStage, groupStage...)

	opts.SetAllowDiskUse(true)

	return collection.Aggregate(ctx, sStage, &opts)
}

// PipeAll executes a MongoDB aggregation query and returns the results
func PipeAll(ctx context.Context, coll string, matchStage bson.D, groupStage ...bson.D) (*[]bson.M, error) {
	var showsWithInfo []bson.M
	cur, err := aggregate(ctx, coll, matchStage, groupStage...)
	if err != nil {
		return nil, err
	}
	defer cur.Close(ctx)

	if err := cur.All(ctx, &showsWithInfo); err != nil {
		g.Log().Error(ctx, "Failed to retrieve documents:", err)
		return nil, err
	}

	return &showsWithInfo, nil
}

// getFromCacheOrAggregate 从缓存获取结果或执行聚合查询
func getFromCacheOrAggregate(ctx context.Context, coll string, matchStage bson.D, groupStage ...bson.D) (*[]bson.M, error) {
	var showsWithInfo []bson.M

	gscache := cache.Instance()
	cacheKey := generateCacheKey(coll, matchStage, groupStage...)

	// 尝试从缓存获取
	resCache, err := gscache.Get(ctx, cacheKey)

	if resCache != nil {
		if err := gconv.Struct(resCache, &showsWithInfo); err != nil {
			g.Log().Warning(ctx, "Failed to unmarshal cache data:", err)
		} else {
			return &showsWithInfo, nil
		}
	}

	// 执行查询
	cur, err := aggregate(ctx, coll, matchStage, groupStage...)
	if err != nil {
		return nil, fmt.Errorf("aggregation failed: %w", err)
	}
	defer cur.Close(ctx)

	if err := cur.All(ctx, &showsWithInfo); err != nil {
		return nil, fmt.Errorf("failed to retrieve all documents: %w", err)
	}

	// 设置缓存
	if len(showsWithInfo) > 0 && !charset.StrContains(gconv.Strings(nocacheList), coll) {
		gscache.SetIfNotExist(ctx, cacheKey, showsWithInfo, global.CacheTime*time.Minute)
	}

	return &showsWithInfo, nil
}

// PipeAllCache executes a MongoDB aggregation query with caching
func PipeAllCache(ctx context.Context, coll string, matchStage bson.D, groupStage ...bson.D) (*[]bson.M, error) {
	return getFromCacheOrAggregate(ctx, coll, matchStage, groupStage...)
}

// PipeIter executes a MongoDB aggregation query and returns a cursor
func PipeIter(ctx context.Context, coll string, matchStage bson.D, groupStage ...bson.D) (*mongo.Cursor, error) {
	return aggregate(ctx, coll, matchStage, groupStage...)
}
